package que.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import que.app.SqlApp;
import que.common.Word;
import que.util.SQLUtil;

import java.time.Duration;

/**
 * @author Naruto
 * @description: 订单与事实表 一条数据一个订单
 * @Class_Name que.app.dwd.db.DwdOrderInfo
 * @Date 2022/09/03 11:28
 */
public class DwdOrderInfo extends SqlApp {
    public static void main(String[] args) {
        new DwdOrderInfo().init(
                9011,
                2,
                "DwdOrderInfo"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        //十秒 超过10s就join不上
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30));
        readOdsDb(tEnv,"DwdOrderInfo");

        //读取字典表
//        readBaseDic(tEnv);
        //一个订单
        Table orderInfo = tEnv.sqlQuery(
                "select " +
                        "data['id'] id, " +
                        "data['user_id'] user_id, " +
                        "data['final_amount'] final_amount, " +
                        "ts, " +
                        "pt " +
                        "from ods_db " +
                        "where `database` = 'edu' " +
                        "and `table` = 'order_info' " +
                        "and `type` = 'insert'"
        );

        tEnv.createTemporaryView("order_info",orderInfo);



        // 读取page页面下单数据
        tEnv.executeSql("create table page_info(" +
                " `page` map<string, string>, " +
                " `common` map<string, string>, " +
                " `ts` bigint, " +
                " `pt` as proctime() " +  // lookup join的时候要用
                ")" + SQLUtil.getKafkaSource(Word.TOPIC_DWD_TRAFFIC_PAGE, "DwdOrderInfo"));

        Table pageOrder = tEnv.sqlQuery(
                " select " +
                        " common['uid'] uid, " +
                        " common['sc'] sc, " +
                        " page['item'] order_id " +
                        " from page_info " +
                        " where " +
                        " page['page_id'] = 'order' " +
                        " and page['item_type'] = 'order_id'"
        );
        tEnv.createTemporaryView("page_Order",pageOrder);


        Table result = tEnv.sqlQuery(
                "select " +
                        " cast( o.id as string) id," +
                        " cast( o.user_id as string) user_id , " +
                        " cast(o.final_amount as string) final_amount," +
                        " p.sc," +
//                        " dic.source_site," +
                        " o.ts," +
                        " o.pt " +
                        " from order_info o  join page_Order p " +
                        "       on o.id = p.order_id and o.user_id = p.uid"
//                      +  "  join base_source for system_time as of o.pt as dic " +
//                        "   on dic.id = p.sc"
        );
//        result.execute().print();
        //定义一个动态表语kafka的topic关联
        tEnv.executeSql(
                "create table DWD_Order_Source_Info(" +
                        " id string," +
                        " user_id string, " +
                        " final_amount string," +
                        " sc string," +
//                        " source_site string," +
                        " ts bigint," +
                        " pt TIMESTAMP_LTZ(3) " +
                        ")" + SQLUtil.getKafkaSink(Word.TOPIC_DWD_ORDER_SOURCE_INFO)
        );


        result.executeInsert("DWD_Order_Source_Info");
    }
}
